package com.lvmama.java.rhino.spark.kafka.consumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class KafkaMessageConsumer {
	
	private List<KafkaMessageProcesser> processerList = new ArrayList<KafkaMessageProcesser>();
	
	public void processMessage(Map<String, Map<Integer, String>> msgs) {
		for (Map.Entry<String, Map<Integer, String>> entry : msgs.entrySet()) {
			System.out.println("Consumer Message received: ");
			System.out.println(("Suchit Topic:" + entry.getKey()));
			for (String msg : entry.getValue().values()) {
				System.out.println("Suchit Consumed Message: " + msg);
				for (KafkaMessageProcesser processer : processerList) {
					processer.process(msg);
				}
			}
		}
	}
}
